package com.wande.dataplatform.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.wande.dataplatform.common.constant.DataPlatformConstants;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.common.exception.DataPlatformException;
import com.wande.dataplatform.common.utils.DorisConnectionManager;
import com.wande.dataplatform.domain.DataModel;
import com.wande.dataplatform.service.IDorisService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Doris 数据仓库服务实现
 * 
 * @author ruoyi
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DorisServiceImpl implements IDorisService {

    private final DorisConnectionManager connectionManager;

    private static final String TENANT_DB_PREFIX = "tenant_";

    @Override
    public void createDatabase(String dbName) {
        if (StrUtil.isBlank(dbName)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "数据库名称不能为空");
        }

        Connection conn = null;
        Statement stmt = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();

            String sql = String.format("CREATE DATABASE IF NOT EXISTS `%s`", dbName);
            stmt.execute(sql);

            log.info("成功创建 Doris 数据库: {}", dbName);

        } catch (SQLException e) {
            log.error("创建 Doris 数据库失败: {}", dbName, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_DDL_FAILED, "创建数据库失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(null, stmt, conn);
        }
    }

    @Override
    public void dropDatabase(String dbName) {
        if (StrUtil.isBlank(dbName)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "数据库名称不能为空");
        }

        Connection conn = null;
        Statement stmt = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();

            String sql = String.format("DROP DATABASE IF EXISTS `%s`", dbName);
            stmt.execute(sql);

            log.info("成功删除 Doris 数据库: {}", dbName);

        } catch (SQLException e) {
            log.error("删除 Doris 数据库失败: {}", dbName, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_DDL_FAILED, "删除数据库失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(null, stmt, conn);
        }
    }

    @Override
    public boolean databaseExists(String dbName) {
        List<String> databases = listDatabases();
        return databases.contains(dbName);
    }

    @Override
    public List<String> listDatabases() {
        List<String> databases = new ArrayList<>();
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            rs = stmt.executeQuery("SHOW DATABASES");

            while (rs.next()) {
                databases.add(rs.getString(1));
            }

            log.debug("查询到 {} 个数据库", databases.size());

        } catch (SQLException e) {
            log.error("查询数据库列表失败", e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "查询数据库列表失败");
        } finally {
            connectionManager.closeResources(rs, stmt, conn);
        }

        return databases;
    }

    @Override
    public void createTable(String dbName, DataModel dataModel) {
        if (StrUtil.isBlank(dbName) || dataModel == null) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "参数不能为空");
        }

        if (!databaseExists(dbName)) {
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_DATABASE_NOT_FOUND, "数据库不存在: " + dbName);
        }

        String ddl = generateCreateTableDDL(dbName, dataModel);

        Connection conn = null;
        Statement stmt = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            stmt.execute(ddl);

            log.info("成功创建 Doris 表: {}.{}", dbName, dataModel.getTableName());

        } catch (SQLException e) {
            log.error("创建 Doris 表失败: {}.{}", dbName, dataModel.getTableName(), e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_DDL_FAILED, "创建表失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(null, stmt, conn);
        }
    }

    @Override
    public void dropTable(String dbName, String tableName) {
        if (StrUtil.isBlank(dbName) || StrUtil.isBlank(tableName)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "参数不能为空");
        }

        Connection conn = null;
        Statement stmt = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();

            String sql = String.format("DROP TABLE IF EXISTS `%s`.`%s`", dbName, tableName);
            stmt.execute(sql);

            log.info("成功删除 Doris 表: {}.{}", dbName, tableName);

        } catch (SQLException e) {
            log.error("删除 Doris 表失败: {}.{}", dbName, tableName, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_DDL_FAILED, "删除表失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(null, stmt, conn);
        }
    }

    @Override
    public boolean tableExists(String dbName, String tableName) {
        List<String> tables = listTables(dbName);
        return tables.contains(tableName);
    }

    @Override
    public List<String> listTables(String dbName) {
        if (StrUtil.isBlank(dbName)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "数据库名称不能为空");
        }

        List<String> tables = new ArrayList<>();
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();

            String sql = String.format("SHOW TABLES FROM `%s`", dbName);
            rs = stmt.executeQuery(sql);

            while (rs.next()) {
                tables.add(rs.getString(1));
            }

            log.debug("数据库 {} 中有 {} 个表", dbName, tables.size());

        } catch (SQLException e) {
            log.error("查询表列表失败: {}", dbName, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "查询表列表失败");
        } finally {
            connectionManager.closeResources(rs, stmt, conn);
        }

        return tables;
    }

    @Override
    public Map<String, Object> getTableSchema(String dbName, String tableName) {
        if (StrUtil.isBlank(dbName) || StrUtil.isBlank(tableName)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "参数不能为空");
        }

        Map<String, Object> schema = new HashMap<>();
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();

            String sql = String.format("DESC `%s`.`%s`", dbName, tableName);
            rs = stmt.executeQuery(sql);

            List<Map<String, String>> columns = new ArrayList<>();
            while (rs.next()) {
                Map<String, String> column = new HashMap<>();
                column.put("field", rs.getString("Field"));
                column.put("type", rs.getString("Type"));
                column.put("null", rs.getString("Null"));
                column.put("key", rs.getString("Key"));
                column.put("default", rs.getString("Default"));
                column.put("extra", rs.getString("Extra"));
                columns.add(column);
            }

            schema.put("database", dbName);
            schema.put("table", tableName);
            schema.put("columns", columns);

            log.debug("获取表结构: {}.{}, 字段数: {}", dbName, tableName, columns.size());

        } catch (SQLException e) {
            log.error("获取表结构失败: {}.{}", dbName, tableName, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "获取表结构失败");
        } finally {
            connectionManager.closeResources(rs, stmt, conn);
        }

        return schema;
    }

    @Override
    public String createTenantDatabase(Long tenantId) {
        if (tenantId == null) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "租户ID不能为空");
        }

        String dbName = getTenantDatabaseName(tenantId);

        if (!databaseExists(dbName)) {
            createDatabase(dbName);
            log.info("为租户 {} 创建数据库: {}", tenantId, dbName);
        } else {
            log.info("租户 {} 的数据库已存在: {}", tenantId, dbName);
        }

        return dbName;
    }

    @Override
    public String getTenantDatabaseName(Long tenantId) {
        if (tenantId == null) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "租户ID不能为空");
        }
        return TENANT_DB_PREFIX + tenantId;
    }

    @Override
    public boolean testConnection() {
        return connectionManager.testConnection();
    }

    @Override
    public List<Map<String, Object>> executeQuery(String sql) {
        if (StrUtil.isBlank(sql)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "SQL不能为空");
        }

        List<Map<String, Object>> results = new ArrayList<>();
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);

            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();

            while (rs.next()) {
                Map<String, Object> row = new HashMap<>();
                for (int i = 1; i <= columnCount; i++) {
                    row.put(metaData.getColumnName(i), rs.getObject(i));
                }
                results.add(row);
            }

            log.debug("执行查询SQL成功，返回 {} 行数据", results.size());

        } catch (SQLException e) {
            log.error("执行查询SQL失败: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "执行查询失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(rs, stmt, conn);
        }

        return results;
    }

    @Override
    public int executeUpdate(String sql) {
        if (StrUtil.isBlank(sql)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "SQL不能为空");
        }

        Connection conn = null;
        Statement stmt = null;
        int affectedRows = 0;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            affectedRows = stmt.executeUpdate(sql);

            log.debug("执行更新SQL成功，影响 {} 行数据", affectedRows);

        } catch (SQLException e) {
            log.error("执行更新SQL失败: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "执行更新失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(null, stmt, conn);
        }

        return affectedRows;
    }

    /**
     * 生成创建表的 DDL 语句
     * 
     * @param dbName 数据库名称
     * @param dataModel 数据模型
     * @return DDL 语句
     */
    private String generateCreateTableDDL(String dbName, DataModel dataModel) {
        StringBuilder ddl = new StringBuilder();
        ddl.append("CREATE TABLE IF NOT EXISTS `").append(dbName).append("`.`").append(dataModel.getTableName()).append("` (\n");

        // 解析字段定义
        JSONArray fields = JSONUtil.parseArray(dataModel.getFields());
        List<String> keyColumns = new ArrayList<>();
        List<String> aggregateColumns = new ArrayList<>();

        for (int i = 0; i < fields.size(); i++) {
            JSONObject field = fields.getJSONObject(i);
            String fieldName = field.getStr("name");
            String fieldType = field.getStr("type");
            Integer length = field.getInt("length");
            Integer scale = field.getInt("scale");
            String comment = field.getStr("comment", "");
            Boolean primaryKey = field.getBool("primaryKey", false);
            Boolean nullable = field.getBool("nullable", true);
            String defaultValue = field.getStr("defaultValue");
            String aggregateType = field.getStr("aggregateType");

            // 主键字段
            if (Boolean.TRUE.equals(primaryKey)) {
                keyColumns.add(fieldName);
            }

            // 构建字段定义
            ddl.append("  `").append(fieldName).append("` ");
            
            // 字段类型和长度
            String fullType = buildFieldType(fieldType, length, scale);
            ddl.append(fullType);

            // 聚合类型(仅用于AGGREGATE表的非主键字段)
            if (StrUtil.isNotBlank(aggregateType) && !Boolean.TRUE.equals(primaryKey)) {
                ddl.append(" ").append(aggregateType);
                aggregateColumns.add(fieldName);
            }

            // NULL约束
            if (!Boolean.TRUE.equals(nullable)) {
                ddl.append(" NOT NULL");
            }

            // 默认值
            if (StrUtil.isNotBlank(defaultValue)) {
                ddl.append(" DEFAULT ").append(defaultValue);
            }

            // 注释
            if (StrUtil.isNotBlank(comment)) {
                ddl.append(" COMMENT '").append(comment.replace("'", "\\'")).append("'");
            }

            if (i < fields.size() - 1) {
                ddl.append(",\n");
            }
        }

        ddl.append("\n)");

        // 表类型和键
        String tableType = dataModel.getTableType();
        if (StrUtil.isBlank(tableType)) {
            tableType = DataPlatformConstants.TableType.DUPLICATE;
        }

        if (DataPlatformConstants.TableType.DUPLICATE.equals(tableType)) {
            ddl.append(" DUPLICATE KEY(");
            // DUPLICATE表如果没有指定主键，使用第一个字段
            if (keyColumns.isEmpty() && fields.size() > 0) {
                keyColumns.add(fields.getJSONObject(0).getStr("name"));
            }
        } else if (DataPlatformConstants.TableType.AGGREGATE.equals(tableType)) {
            ddl.append(" AGGREGATE KEY(");
        } else if (DataPlatformConstants.TableType.UNIQUE.equals(tableType)) {
            ddl.append(" UNIQUE KEY(");
        }

        // 添加键列
        if (!keyColumns.isEmpty()) {
            ddl.append(String.join(", ", keyColumns.stream().map(k -> "`" + k + "`").toArray(String[]::new)));
        }
        ddl.append(")");

        // 分区配置
        if (StrUtil.isNotBlank(dataModel.getPartitionColumn())) {
            ddl.append("\nPARTITION BY RANGE(`").append(dataModel.getPartitionColumn()).append("`)()");
        }

        // 分桶配置
        String bucketColumn = dataModel.getBucketColumn();
        if (StrUtil.isBlank(bucketColumn) && !keyColumns.isEmpty()) {
            // 如果没有指定分桶字段，使用第一个主键字段
            bucketColumn = keyColumns.get(0);
        }
        
        if (StrUtil.isNotBlank(bucketColumn)) {
            ddl.append("\nDISTRIBUTED BY HASH(`").append(bucketColumn).append("`)");
            Integer bucketNum = dataModel.getBucketNum();
            if (bucketNum == null || bucketNum <= 0) {
                bucketNum = 10; // 默认10个桶
            }
            ddl.append(" BUCKETS ").append(bucketNum);
        }

        // 表属性
        ddl.append("\nPROPERTIES (\n");
        ddl.append("  \"replication_num\" = \"1\"\n");
        ddl.append(")");

        log.debug("生成 DDL: {}", ddl);

        return ddl.toString();
    }

    /**
     * 构建字段类型
     */
    private String buildFieldType(String type, Integer length, Integer scale) {
        if (type == null) {
            return "VARCHAR(255)";
        }

        String upperType = type.toUpperCase();
        
        // 需要长度的类型
        if (upperType.equals("VARCHAR") || upperType.equals("CHAR")) {
            int len = (length != null && length > 0) ? length : 255;
            return upperType + "(" + len + ")";
        }
        
        // DECIMAL类型需要精度和标度
        if (upperType.equals("DECIMAL")) {
            int precision = (length != null && length > 0) ? length : 10;
            int scaleVal = (scale != null && scale >= 0) ? scale : 0;
            return "DECIMAL(" + precision + "," + scaleVal + ")";
        }
        
        // 其他类型直接返回
        return upperType;
    }

}
